package cn._51doit.day09;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @create: 2021-10-26 22:37
 * @author: 今晚打脑斧先森
 * @program: SideOutputDemo
 * @Description:
 *    SideOutput（测流输出、旁路输出）
 *    将整个数量流按照指定类型，打上一到多个标签，然后按照指定的标签将数据筛选出来，再对数据进行处理
 *
 *    SideOutput VS Filter (测流输出与过滤的区别)
 *    SideOutput底层是给数据关联上标签，然后根据标签获取指定类型的数据
 *    Filter是将不符合条件的数据过滤掉，保留符合条件的数据，如果想要保留多种类型的数据，需要过滤多次，过滤效果低
 *
 *    需求：将数据中的偶数、奇数、和字符串打上标签
 **/
public class SideOutputDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        OutputTag<Integer> jishu = new OutputTag<Integer>("整数") {};
        OutputTag<Integer> oushu = new OutputTag<Integer>("偶数"){};
        OutputTag<String> str = new OutputTag<String>("字符串") {};
        //主流中只有两种数据,打过标签对的数据和未打标签的数据,ctx输出的数据是打标签的数据,out输出的数据是未打标签的数据
        SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String in, Context ctx, Collector<String> out) throws Exception {
                try {
                    int num = Integer.parseInt(in);
                    if (num % 2 == 0) {
                        //偶数,打上偶数的标签
                        ctx.output(oushu, num);
                    } else //奇数,打上奇数的标签
                        ctx.output(jishu, num);
                } catch (Exception e) {
                    //剩下的就是字符串
                    ctx.output(str, in);
                }
                //输出未打标签的数据
                out.collect(in);
            }
        });
        //从主流中输出打标签的数据  偶数
        DataStream<Integer> sideOutput = mainStream.getSideOutput(oushu);
        sideOutput.print("偶数标签: ");
        //从主流中输出打标签的数据 字符串
        DataStream<String> sideOutput1 = mainStream.getSideOutput(str);
        sideOutput1.print("字符串标签: ");

        //从主流中输出未打标签的数据
        mainStream.print("主流:");
        env.execute();
    }
}
